﻿//Copyright (c) Microsoft Corporation.  All rights reserved.

using System;
using System.Collections.Generic;
using System.Text;

namespace Microsoft.Ted.Wacel.Base
{
    public class CompressedCircularBuffer : CircularBuffer<byte[]>
    {
        public int BatchSize { get; set; }
        public bool AvoidDuplicatedRead { get; set; }
        private long mSize;
        public CompressedCircularBuffer(string name, long size, IDataProvider dataProvider, ILockProvider lockProvider, bool avoidDuplicatedRead = true)
            : base(name, size, dataProvider, lockProvider)
        {
            BatchSize = dataProvider.Configuration.WriteBatchSize; //TODO: is this correct?
            mSize = size;
            AvoidDuplicatedRead = avoidDuplicatedRead;
        }

        public void Add(string[] items)
        {
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < items.Length; i++)
                sb.Append(items[i].Length).Append(";");
            sb.Append("*");
            for (int i = 0; i < items.Length; i++)
                sb.Append(items[i]);
            base.Add(StringExtensions.Zip(Encoding.UTF8.GetBytes(sb.ToString())));
        }
        private int lastHash = 0;
        public string[] Get(long key)
        {
            string load = Encoding.UTF8.GetString(StringExtensions.Unzip(base[key]));
            int hash = AvoidDuplicatedRead ? load.GetHashCode() : 0;
            if (!AvoidDuplicatedRead || hash != lastHash)
            {
                int index = load.IndexOf('*');
                string slices = load.Substring(0, index);
                string[] lengths = slices.Split(';');
                string[] ret = new string[lengths.Length - 1]; //The last element will be between last ';' and the '*' mark, hence empty.
                int scan = index + 1;
                for (int i = 0; i < ret.Length; i++)
                {
                    int size = int.Parse(lengths[i]);
                    ret[i] = load.Substring(scan, size);
                    scan += size;
                }
                lastHash = hash;
                return ret;
            }
            else
                return new string[] { };
        }
        public override void Add(byte[] item)
        {
            Add(new string[] { Encoding.UTF8.GetString(item) });
        }
        private List<string> mInBuffer = new List<string>();
        public  void Add(string item)
        {
            lock (mInBuffer)
            {
                mInBuffer.Add(item);
                if (mInBuffer.Count >= BatchSize)
                {
                    Add(mInBuffer.ToArray());
                    mInBuffer.Clear();
                }
            }
        }

        public void Flush()
        {
            lock (mInBuffer)
            {
                if (mInBuffer.Count > 0)
                {
                    Add(mInBuffer.ToArray());
                    mInBuffer.Clear();
                }
            }
        }
        
        public new string this[long key]
        {
            get
            {
                if (key > 0 || key <= -mSize * BatchSize)
                    throw new IndexOutOfRangeException();
                if (AvoidDuplicatedRead && BatchSize > 1)
                    throw new InvalidOperationException("Indexing when batch is larger than 1 and duplication detection is eanbled is not allowed.");
                long batch = key / BatchSize;
                string[] items = Get(batch);
                long inner = (BatchSize-1)+(key + batch * BatchSize);
                return items[inner];
            }
        }
    }
}
